package com.dx.chat.kafka.service.impl;

import com.dx.chat.kafka.service.IKafkaProducerService;
import org.apache.kafka.common.PartitionInfo;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.function.Function;

/**
 * 生产者
 *
 * @author xufeng
 * @since 13.04.2023
 */
@Service("iKafkaProducerService")
public class KafkaProducerDataGroupServiceImpl<T> implements IKafkaProducerService<T> {

    @Resource
    private KafkaTemplate<String, T> kafkaTemplate;

    @Override
    public void sendMessage(String topic, T message) {
        kafkaTemplate.send(topic, message);
    }

    @Override
    public void sendMessage(String topic, T message, Function function, Function error) {
        kafkaTemplate.send(topic, message).addCallback(function::apply, error::apply);
    }

    @Override
    public void sendMessage(String topic, String key, T message, Function function, Function error) {
        kafkaTemplate.send(topic, key, message).addCallback(function::apply, error::apply);

    }

    @Override
    public void sendMessage(String topic, String key, T message) {
        kafkaTemplate.send(topic, key, message);
    }

    @Override
    public void sendMessage(String topic, Integer partition, String key, T message) {
        kafkaTemplate.send(topic, partition, key, message);

    }

    @Override
    public List<PartitionInfo> partitionsFor(String topic) {
        List<PartitionInfo> partitionInfos = kafkaTemplate.partitionsFor(topic);
        return partitionInfos;
    }
}